package com.bd08.stock.flink

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment;

/**
 * 需要解决 sql 语法错误
 * jar 依赖
 */
object Driver extends App {


  val env = StreamExecutionEnvironment.getExecutionEnvironment
  val tEnv = StreamTableEnvironment.create(env)

  // kafka avro source
  tEnv.executeSql(
    """
      |create table stock_it (
      |refresh_time varchar(50),
      |symbol varchar(50),
      |company_name varchar(50),
      |exchange varchar(50),
      |open_price double,
      |ask_price double,
      |ask_size int,
      |bid_price double,
      |bid_size int,
      |price double
      |)
      |with(
      |'connector' = 'kafka',
      |'topic'     = 'stock_it',
      |'properties.bootstrap.servers' = 'localhost:9092',
      |'properties.group.id' = 'stock_it',
      |'scan.startup.mode' = 'earliest-offset',
      |'value.format' = 'avro-confluent',
      |'value.avro-confluent.url' = 'http://localhost:8081'
      |)
      |""".stripMargin)

  //kafka avro source
  tEnv.executeSql(
    """
      |create table stock_bk (
      |refresh_time varchar(50),
      |symbol varchar(50),
      |company_name varchar(50),
      |exchange varchar(50),
      |open_price double,
      |ask_price double,
      |ask_size int,
      |bid_price double,
      |bid_size int,
      |price double
      |)
      |with(
      |'connector' = 'kafka',
      |'topic'     = 'stock_bk',
      |'properties.bootstrap.servers' = 'localhost:9092',
      |'properties.group.id' = 'stock_bk',
      |'scan.startup.mode' = 'earliest-offset',
      |'value.format' = 'avro-confluent',
      |'value.avro-confluent.url' = 'http://localhost:8081'
      |)
      |""".stripMargin)

  // jdbc sink
  tEnv.executeSql(
    """
      |create table tb_stock(
      |refresh_time varchar(50),
      |symbol varchar(50),
      |company_name varchar(50),
      |exchange varchar(50),
      |open_price double,
      |ask_price double,
      |ask_size int,
      |bid_price double,
      |bid_size int,
      |price double
      |)
      |with(
      | 'connector'  = 'jdbc',
      | 'url'        = 'jdbc:mysql://localhost:3306/stock-db',
      | 'table-name' = 'tb_stock',
      | 'driver'     = 'com.mysql.cj.jdbc.Driver',
      | 'username'   = 'root',
      | 'password'   = 'password'
      |)
      |""".stripMargin)

  val stock_it = tEnv.from("stock_it")
  val stock_bk = tEnv.from("stock_bk")
  stock_it.executeInsert("tb_stock")
  println(".................")
  stock_bk.executeInsert("tb_stock")
  //stock_it.execute().print()

  //val ds =tEnv.toDataStream(stock_it).union(tEnv.toDataStream(stock_bk))
  //val schema=Schema.newBuilder()
//    .column("refresh_time", DataTypes.STRING())
//    .column("symbol", DataTypes.STRING())
//    .column("company_name", DataTypes.STRING())
//    .column("exchange", DataTypes.STRING())
//    .column("open_price", DataTypes.DOUBLE())
//    .column("ask_price", DataTypes.DOUBLE())
//    .column("ask_size", DataTypes.INT())
//    .column("bid_price", DataTypes.DOUBLE())
//    .column("bid_size", DataTypes.INT())
//    .column("price", DataTypes.FLOAT()).build()

//  tEnv.createTemporaryView("stock",ds,schema)
//  val stock = tEnv.from("stock")
//  stock_it.select($"*").execute().print()
  //stock.executeInsert("tb_stock")


}
